package com.margo.project.consumer.kafka.producer.controller;

import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.kafka.core.KafkaTemplate;
import org.springframework.web.bind.annotation.RequestMapping;
import org.springframework.web.bind.annotation.RequestMethod;
import org.springframework.web.bind.annotation.RestController;

import com.fasterxml.jackson.core.JsonProcessingException;
import com.fasterxml.jackson.databind.ObjectMapper;
import com.margo.project.consumer.kafka.consumer.model.User;


/**
 * 	Can't convert value of class com.margo.project.consumer.kafka.consumer.model.User to 
 * 	class org.apache.kafka.common.serialization.StringSerializer specified in value.serializer
 * @author fxb20
 *
 */
@RestController
@RequestMapping("/producer")
public class KafkaProducerController {
	
	@Autowired
    private KafkaTemplate<Object, Object> kafkaTemplate;
	
	@Autowired
	private ObjectMapper objectMapper;

	/**
	 * http://localhost:9001/producer/sendKafka
	 * @return
	 * @throws JsonProcessingException 
	 */
    @RequestMapping(value = "/sendKafka", method = RequestMethod.GET)
    public Object sendKafka() throws JsonProcessingException {
    	User user=new User();
    	user.setName("fxb");
    	user.setAge(31);
        kafkaTemplate.send("margoStreamTopic", null, objectMapper.writeValueAsString(user));
        return "ok";
    }
}
